package com.ruoyi.generator.callback;

import cn.hutool.core.annotation.AnnotationUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.ruoyi.generator.annocation.MqttCallbackKey;
import com.ruoyi.generator.domain.MqttProperty;
import com.ruoyi.generator.util.MQTTClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

public class MqttCallbackActuator {

    private static final Map<String, List<MqttCallbackObserver>> map;

    static {
        map = new ConcurrentHashMap<>();
        Map<String, MqttCallbackObserver> beansOfType = SpringUtil.getBeansOfType(MqttCallbackObserver.class);
        for (MqttCallbackObserver observer : beansOfType.values()) {
            Class<? extends MqttCallbackObserver> aClass = observer.getClass();
            if (AnnotationUtil.hasAnnotation(aClass, MqttCallbackKey.class)) {
                String key = AnnotationUtil.getAnnotationValue(aClass, MqttCallbackKey.class).toString();
                if (map.containsKey(key)) {
                    map.get(key).add(observer);
                } else {
                    ArrayList<MqttCallbackObserver> list = new ArrayList<>();
                    list.add(observer);
                    map.put(key, list);
                }
            }
        }

    }

    private static List<MqttCallbackObserver> getList(String key) {
        return map.get(key);
    }

    public static void connectComplete(boolean reconnect, String serverURI, String key) {
        List<MqttCallbackObserver> list = getList(key);
        if (CollUtil.isNotEmpty(list)) {
            for (MqttCallbackObserver observer : list) {
                observer.connectComplete(reconnect, serverURI);
            }
        }
    }

    public static void connectionLost(Throwable cause, String key) {
        List<MqttCallbackObserver> list = getList(key);
        if (CollUtil.isNotEmpty(list)) {
            for (MqttCallbackObserver observer : list) {
                observer.connectionLost(cause);
            }
        }
    }

    public static void messageArrived(String topic, MqttMessage message, String key) throws Exception {
        List<MqttCallbackObserver> list = getList(key);
        if (CollUtil.isNotEmpty(list)) {
            for (MqttCallbackObserver observer : list) {
                observer.messageArrived(topic, message);
            }
        }
    }

    public static void deliveryComplete(IMqttDeliveryToken token, String key) {
        List<MqttCallbackObserver> list = getList(key);
        if (CollUtil.isNotEmpty(list)) {
            for (MqttCallbackObserver observer : list) {
                observer.deliveryComplete(token);
            }
        }
    }


    public static void init(MQTTClient mqttClient) {
        for (Map.Entry<String, List<MqttCallbackObserver>> entry : map.entrySet()) {
            if (entry.getKey().equals(mqttClient.getMqttProperty().getKey()) && (CollUtil.isNotEmpty(entry.getValue()))) {
                for (MqttCallbackObserver observer : entry.getValue()) {
                    observer.setMqttClient(mqttClient);
                }
            }
        }
    }
}
